package com.demo;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

@Deprecated
public class MyRedisSink extends RichSinkFunction<Tuple2<String, Long>> {

  private Jedis jedis;
  private String host;
  private int port;

  public MyRedisSink() {
  }

  public MyRedisSink(String host, int port) {
    this.host = host;
    this.port = port;
  }

  @Override
  public void open(Configuration parameters) throws Exception {
    super.open(parameters);
    this.jedis = new Jedis(host, port);
  }

  @Override
  public void close() throws Exception {
    super.close();
    if (this.jedis != null) {
      this.jedis.close();
    }
  }

  @Override
  public void invoke(Tuple2<String, Long> value, Context context) throws Exception {
    this.jedis.incrBy(value.f0, value.f1);
  }
}
